package yuan.consumer.multy;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import yuan.consumer.KafkaConsumerAnalysis;

import java.util.Arrays;
import java.util.Properties;

/**
 * @version: 1.0
 * @Description: TODO
 * @author: zZmH
 * @date: 2020/09/09 13:29
 */


public class FirstMultiConsumeThreadDemo {

    public static Properties init(){
        Properties properties = new Properties();
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer.client.id.demo");
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);

        return properties;

    }
    static String topic = "topic-demo";
    static String brokerList = "192.168.190.128:9092";
    public static final String groupId = "group.demo";


    public static void main(String[] args) {

        final Properties prop = init();

        for (int i = 0; i < 3; i++) {
            new KafkaConsumerThread(prop, topic).start();
        }



    }

    public static class KafkaConsumerThread extends Thread {


        private KafkaConsumer<String,String> consumer;

        public KafkaConsumerThread(Properties prop,String topic){
            consumer = new KafkaConsumer<String, String>(prop);
            consumer.subscribe(Arrays.asList(topic));

        }
        @Override
        public void run() {


            try {

                while (true) {
                    final ConsumerRecords<String, String> records = consumer.poll(100);
                    for (ConsumerRecord<String, String> record : records) {
                        //do sth
                    }


                }

            } catch (Exception e) {
                e.printStackTrace();
            }finally{
                consumer.close();
            }

        }
    }
}
